package com.源码分析.processor;

import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.Flux;
import reactor.core.publisher.ReplayProcessor;

// 可以缓存通过sink产生的数据或者订阅publisher的数据，然后重放给后来的订阅者。有如下四种配置
// cacheLast
// 只缓存最后一个数据
// create(int)
// 缓存最后N个数据
// createTimeout(Duration)
// 对每个数据打上时间戳标签，只缓存age在指定ttl内的数据
// createSizeOrTimeout(int,Duration)
// 对每个数据打上时间戳标签，只缓存age在指定ttl内的N个数据
public class ReplayProcessorDemo {

	static final Logger LOGGER = LoggerFactory.getLogger(ReplayProcessorDemo.class);

	@Test
    public void testReplayProcessor() throws InterruptedException {
        ReplayProcessor<Integer> replayProcessor = ReplayProcessor.create(3);
        Flux<Integer> flux1 = replayProcessor
                .map(e -> e);
        Flux<Integer> flux2 = replayProcessor
                .map(e -> e);

        flux1.subscribe(e -> {
            LOGGER.info("flux1 subscriber:{}",e);
        });


        IntStream.rangeClosed(1,5)
                .forEach(e -> {
                    replayProcessor.onNext(e);
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });

        LOGGER.info("finish publish data");
        TimeUnit.SECONDS.sleep(3);

        LOGGER.info("begin to subscribe flux2");
        flux2.subscribe(e -> {
            LOGGER.info("flux2 subscriber:{}",e);
        });

        replayProcessor.onComplete();
        replayProcessor.blockLast();
    }

}
